package dal

import (
	"context"
	"database/sql"
	"fmt"
	"time"

	"code.byted.org/bytertc/whiteboard_server/biz/pkg/kv"
	"github.com/spf13/cast"

	"code.byted.org/bytertc/whiteboard_server/biz/dal/ddl"

	v1 "code.byted.org/bytertc/whiteboard_server/biz/api/pb/v1"
	"code.byted.org/bytertc/whiteboard_server/biz/dal/spec"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/biz_error"
	"code.byted.org/bytertc/whiteboard_server/biz/service/action/draw/phase"
	"code.byted.org/gopkg/logs/v2/log"
	"github.com/pkg/errors"
	"google.golang.org/protobuf/proto"
	"gorm.io/plugin/dbresolver"
)

// list, value = binary(v1.Packet)
func genPhasesKey(unique spec.ActionUnique) string {
	return fmt.Sprintf("action:%s:%s:%s:%d",
		unique.GetAppId(), unique.GetRoomId(), unique.GetObjId(), unique.GetActionId())
}

// string, value = sum(packets of a board)
func genSavedPacketIdKey(unique spec.BoardUnique) string {
	return fmt.Sprintf("pktCnt:%s:%s",
		unique.GetAppId(), unique.GetRoomId())
}

func genTransformActionCount(unique spec.ActionUnique, operator string) string {
	return fmt.Sprintf("transfomrCnt:%s:%s:%s:%s:%d",
		unique.GetAppId(), unique.GetRoomId(), operator, unique.GetObjId(), unique.GetActionId())
}

func genCursorStateKey(unique spec.BoardUnique) string {
	return fmt.Sprintf("cursorStateList:%s:%s:%d", unique.GetAppId(), unique.GetRoomId(), unique.GetBoardId())
}

func (r *repo) GetPhases(ctx context.Context, unique spec.ActionUnique) ([]phase.Phase, error) {
	key := genPhasesKey(unique)
	perPage := 10
	rawData := make([]string, 0)
	for i := 0; i < ForMaxCycleNum; i++ {
		start := int64(i * perPage)
		end := int64((i+1)*perPage - 1)
		tmpRawData, err := r.cache.LRange(ctx, key, start, end)
		if err != nil {
			return nil, err
		}
		rawData = append(rawData, tmpRawData...)
		if len(tmpRawData) < perPage {
			break
		}
	}

	phases := make([]phase.Phase, 0)
	for index := range rawData {
		packet := &v1.Packet{}
		err := proto.Unmarshal([]byte(rawData[index]), packet)
		if err != nil {
			return nil, biz_error.DecodeError.Wrap(errors.WithStack(err))
		}
		ps, err := phase.NewDrawPhases(ctx, unique.GetAppId(), unique.GetRoomId(), packet)
		if err != nil {
			return nil, err
		}
		phases = append(phases, ps...)
	}
	log.V1.CtxInfo(ctx, "got %d actions from list %s", len(phases), key)
	return phases, nil
}

func (r *repo) SavePhases(ctx context.Context, phases []phase.Phase) error {
	if len(phases) == 0 {
		return nil
	}
	packet := &v1.Packet{
		BoardId:  phases[0].GetBoardId(),
		PageId:   phases[0].GetPageId(),
		Operator: phases[0].GetOperator(),
		Version:  1,
		Actions:  make([]*v1.Action, 0, len(phases)),
	}
	key := genPhasesKey(phases[0])
	//	二次编辑需要保证ongoing前面有start
	if phases[0].IsModifyOnGoing() {
		length, _ := r.cache.LLen(ctx, key)
		if length == 0 {
			return biz_error.NotFound.Wrap(errors.Errorf("not exist target action start phase key:%s", key))
		}
	}

	for index, p := range phases {
		if p.IsCreateActionId() && p.GetPhase() != v1.PhaseType_kOnGoing {
			continue
		}
		act, err := phases[index].ToProto()
		if err != nil {
			return err
		}
		packet.Actions = append(packet.Actions, act)
	}

	data, err := proto.Marshal(packet)
	if err != nil {
		return biz_error.EncodeError.Wrap(errors.WithStack(err))
	}
	count, err := r.cache.RPush(ctx, key, []string{string(data)})
	_, _ = r.cache.Expire(ctx, key, 2*time.Minute)
	log.V1.CtxInfo(ctx, "%d packets in list", count)
	// todo add metrics
	return err
}

func (r *repo) SavePacket(ctx context.Context, unique spec.BoardUnique, packet *v1.Packet) (int64, int64, error) {
	data, err := proto.Marshal(&v1.PlaybackFile{Packets: []*v1.Packet{packet}}) // where magic happens
	if err != nil {
		return 0, 0, biz_error.EncodeError.Wrap(errors.WithStack(err))
	}
	actionCount := int32(len(packet.Actions))

	rawPacket := &ddl.Packet{
		AppId:       unique.GetAppId(),
		RoomId:      unique.GetRoomId(),
		BoardId:     packet.BoardId,
		ActionCount: actionCount,
		StartTime:   packet.Actions[0].Timestamp,
		Version:     1,
		Data:        data,
	}

	err = r.db.WithContext(ctx).Table(ddl.TablePackets).Create(rawPacket).Error
	if err != nil {
		return 0, 0, biz_error.DatabaseError.Wrap(errors.WithStack(err))
	}

	packetCount, err := r.cache.IncrBy(ctx, genSavedPacketIdKey(unique), 1)
	if err != nil {
		return rawPacket.Id, 0, biz_error.CacheError.Wrap(errors.WithStack(err))
	}

	return rawPacket.Id, packetCount, nil
}

func (r *repo) GetPackets(ctx context.Context, unique spec.RoomUnique, maxId int64) ([]*ddl.Packet, error) {
	var ps []ddl.Packet
	var lowerBound sql.NullInt64

	err := r.db.WithContext(ctx).Clauses(dbresolver.Write).Table(ddl.TableReplays).
		Where(`app_id = ? and room_id = ?`,
			unique.GetAppId(), unique.GetRoomId()).
		Select("max(max_id)").Row().Scan(&lowerBound)
	if err != nil {
		return nil, biz_error.DatabaseError.Wrap(errors.WithStack(err))
	}
	// 单房间高QPS情况下可能出现两个 maxId 基于同一个 lowerBound 的情况 (maxId2 - maxId1 = n*packetPerFile)

	err = r.db.WithContext(ctx).Clauses(dbresolver.Write).Table(ddl.TablePackets).
		Where("app_id = ? and room_id = ? and id > ? and id <= ?",
			unique.GetAppId(), unique.GetRoomId(), lowerBound.Int64, maxId).
		Order("id").Find(&ps).Error
	if err != nil {
		return nil, biz_error.DatabaseError.Wrap(errors.WithStack(err))
	}

	packets := make([]*ddl.Packet, len(ps))
	for index := range ps {
		packets[index] = &ps[index]
	}
	return packets, nil
}
func (r *repo) GetMaxPacketId(ctx context.Context, unique spec.RoomUnique) (int64, error) {
	var maxId sql.NullInt64

	err := r.db.WithContext(ctx).Clauses(dbresolver.Write).Table(ddl.TablePackets).
		Where(`app_id = ? and room_id = ?`,
			unique.GetAppId(), unique.GetRoomId()).
		Select("max(id)").Row().Scan(&maxId)
	if err != nil {
		return 0, biz_error.DatabaseError.Wrap(errors.WithStack(err))
	}
	return maxId.Int64, nil
}

func (r *repo) SaveReplay(ctx context.Context, replay *ddl.Replay) error {
	err := r.db.WithContext(ctx).Table(ddl.TableReplays).Create(replay).Error
	return biz_error.DatabaseError.Wrap(err)
}

func (r *repo) ListReplay(ctx context.Context, unique spec.BoardUnique) ([]ddl.Replay, error) {
	var files []ddl.Replay
	err := r.db.WithContext(ctx).Table(ddl.TableReplays).
		Where("app_id = ? and room_id = ?",
			unique.GetAppId(), unique.GetRoomId()).
		Order("id").Find(&files).Error
	return files, biz_error.DatabaseError.Wrap(errors.WithStack(err))
}

func (r *repo) IncrTransformCount(ctx context.Context, unique spec.ActionUnique, operator string) (int64, error) {
	cacheKey := genTransformActionCount(unique, operator)
	count, err := r.cache.IncrBy(ctx, cacheKey, 1)
	if err != nil {
		return 0, biz_error.CacheError.Wrap(errors.WithStack(err))
	}
	_, err = r.cache.Expire(ctx, cacheKey, time.Second*10)
	if err != nil {
		log.V1.CtxError(ctx, "expire fail key:%s,err:%+v", cacheKey, err)
	}
	return count, nil

}

func (r *repo) ChangeCursorState(ctx context.Context, unique spec.BoardUnique, operator string, state *v1.ChangeCursorState) error {
	cacheKey := genCursorStateKey(unique)
	if state.ShowState == v1.UserCursorInfo_kHide {
		err := r.cache.ZRem(ctx, cacheKey, operator)
		if err != nil {
			log.V1.CtxError(ctx, "del fail key:%s,user:%s,err:%+v", cacheKey, operator, err)
		}
	} else {
		err := r.cache.ZAdd(ctx, cacheKey, kv.ZMember{
			Member: operator,
			Score:  float64(state.CursorType),
		})
		if err != nil {
			log.V1.CtxError(ctx, "zadd fail key:%s,user:%s,err:%+v", cacheKey, operator, err)
			return biz_error.CacheError.Wrap(err)
		}
		_, err = r.cache.Expire(ctx, cacheKey, time.Hour*12)
		if err != nil {
			log.V1.CtxError(ctx, "Expire fail key:%s,user:%s,err:%+v", cacheKey, operator, err)
		}
	}
	return nil
}

func (r *repo) GetTopNCursorState(ctx context.Context, unique spec.BoardUnique) (map[string]*v1.UserCursorInfo, error) {
	cacheKey := genCursorStateKey(unique)
	members, err := r.cache.ZRangeWithScore(ctx, cacheKey, 0, 19)
	if err != nil {
		log.V1.CtxError(ctx, "zrange fail key:%s,err:%+v", cacheKey, err)
		return nil, biz_error.CacheError.Wrap(err)
	}
	ret := make(map[string]*v1.UserCursorInfo)
	for _, member := range members {
		info := &v1.UserCursorInfo{
			ShowState:  v1.UserCursorInfo_kShow,
			CursorType: v1.UserCursorInfo_CursorType(member.Score),
		}
		ret[cast.ToString(member.Member)] = info
	}
	return ret, nil

}

func (r *repo) GetMyCursorState(ctx context.Context, unique spec.BoardUnique, operator string) (*v1.ChangeCursorState, error) {
	cacheKey := genCursorStateKey(unique)
	score, err := r.cache.ZScore(ctx, cacheKey, operator)
	if err != nil && !IsNotFoundError(err) {
		return nil, biz_error.CacheError.Wrap(err)
	}
	if IsNotFoundError(err) {
		return &v1.ChangeCursorState{
			ShowState:  v1.UserCursorInfo_kHide,
			CursorType: v1.UserCursorInfo_CursorType(score),
		}, nil
	}
	return &v1.ChangeCursorState{
		ShowState:  v1.UserCursorInfo_kShow,
		CursorType: v1.UserCursorInfo_CursorType(score),
	}, nil

}
